package com.cctc.rds.raft.core.rpc.log.impl;

import com.cctc.rds.raft.core.RaftCluster;
import com.cctc.rds.raft.core.log.Log;
import com.cctc.rds.raft.core.log.LogService;
import com.cctc.rds.raft.core.log.commit.LogCommitter;
import com.cctc.rds.raft.core.log.commit.statemachine.StateMachine;
import com.cctc.rds.raft.core.node.Node;
import com.cctc.rds.raft.core.rpc.heartbeat.HeartBeatListener;
import com.cctc.rds.raft.core.rpc.log.LogRpc;
import com.cctc.rds.raft.core.rpc.log.LogRpcRequest;
import com.cctc.rds.raft.core.rpc.log.LogRpcResponse;

public class LogRpcResponseImpl implements LogRpc {
    private int nodeId;
    private LogService logService;
    private LogCommitter logApply;

    public LogRpcResponseImpl(int nodeId, LogService logService, StateMachine stateMachine) {
        this.nodeId = nodeId;
        this.logService = logService;
        this.logApply = new LogCommitter(logService, stateMachine);
    }

    @Override
    public LogRpcResponse append(LogRpcRequest request) {
        resetElectionRpcScheduler();
        LogRpcResponse response = LogRpcResponse.builder().build();
        Node curNode = RaftCluster.getNode(nodeId);
        if (curNode.getCurTerm() > request.getTerm()) {
            response.setSuccess(false);
            response.setTerm(curNode.getCurTerm());
            return response;
        }
        Log log = logService.index(request.getLastCommitLogIndex());
        if ((log == null && request.getLastCommitLogTerm() != -1)
                || (log != null && request.getLastCommitLogTerm() != log.getTerm())) {
            response.setSuccess(false);
            response.setTerm(request.getTerm());
            return response;
        }
        if (request.getLogs() == null || request.getLogs().length == 0) {
            RaftCluster.getNode(request.getNodeId()).asLeader(request.getTerm());
            response.setSuccess(true);
            response.setTerm(request.getTerm());
            return response;
        }
        Log[] requestLogs = request.getLogs();
        int notExistLogIndex = 0;
        for (int i = 0; i < requestLogs.length; i++) {
            Log requestLog = requestLogs[i];
            log = logService.index(requestLog.getIndex());
            if (log != null && log.getTerm() != requestLog.getTerm()) {
                logService.remove(log.getIndex(), logService.peek().getIndex());
                notExistLogIndex = i;
                break;
            }
        }
        while (notExistLogIndex < requestLogs.length) {
            Log appendLog = requestLogs[notExistLogIndex];
            logService.append(appendLog);
            if (appendLog.getStatus() == 1) logApply.commit(appendLog);
            notExistLogIndex++;
        }
        response.setSuccess(true);
        response.setTerm(request.getTerm());
        if (curNode.curCommitLogIndex() < request.getFollowerCommitLogIndex()) {
            curNode.seekCommitLogIndex(request.getFollowerCommitLogIndex());
        }
        return response;
    }

    private void resetElectionRpcScheduler() {
        try {
            HeartBeatListener listener = RaftCluster.getHeartBeatListener();
            if (listener != null) {
                listener.onHeartBeat();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public Boolean commit(long index, long term) {
        Log log = logService.index(index);
        log.setTerm(term);
        boolean flag = logApply.commit(log);
        if (flag)
            RaftCluster.getNode(nodeId).seekCommitLogIndex(log.getIndex());
        return flag;
    }
}
